package com.rosh.rpc.client.request;

import com.rosh.rpc.client.cache.ServiceProviderCache;
import com.rosh.rpc.core.data.RpcRequest;
import com.rosh.rpc.core.data.RpcResponse;
import com.rosh.rpc.core.netty.codec.FrameDecoder;
import com.rosh.rpc.core.netty.codec.FrameEncoder;
import com.rosh.rpc.core.netty.codec.RpcRequestEncoder;
import com.rosh.rpc.core.netty.codec.RpcResponseDecoder;
import com.rosh.rpc.core.netty.handler.RpcResponseHandler;
import com.rosh.rpc.core.netty.request.ChannelMapping;
import com.rosh.rpc.core.netty.request.RpcRequestHolder;
import com.rosh.rpc.core.provider.ServiceProvider;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.DefaultPromise;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @Description:
 * @Author: rosh
 * @Date: 2022/6/14 22:10
 */
@Component
@Slf4j
public class RpcRequestManager {

    @Autowired
    private ServiceProviderCache providerCache;


    public RpcResponse sendRequest(RpcRequest request) {
        // 1、获取可用发服务节点
        List<ServiceProvider> serviceProviders = providerCache.get(request.getClassName());
        if (CollectionUtils.isEmpty(serviceProviders)) {
            log.info("客户端没有发现可用发服务节点");
            throw new IllegalArgumentException("客户端没有发现可用发服务节点");
        }
        ServiceProvider serviceProvider = serviceProviders.get(0);
        return requestByNetty(request, serviceProvider);
    }

    private RpcResponse requestByNetty(RpcRequest request, ServiceProvider serviceProvider) {
        try {
            //channel 复用
            if (!RpcRequestHolder.channelExist(serviceProvider.getServerIp(), serviceProvider.getRpcPort())) {
                createChannel(serviceProvider);
            }
            Channel channel = RpcRequestHolder.getChannel(serviceProvider.getServerIp(), serviceProvider.getRpcPort());
            if (channel != null) {
                //向服务端发送数据
                DefaultPromise<RpcResponse> promise = new DefaultPromise<>(channel.eventLoop());
                //建立映射
                RpcRequestHolder.addRequestPromise(request.getRequestId(), promise);
                //发送数据
                channel.writeAndFlush(request);
                //监听
                RpcResponse rpcResponse = promise.get();
                //监听成功后移除
                RpcRequestHolder.removeRequestPromise(request.getRequestId());
                return rpcResponse;
            }
        } catch (Exception e) {
            log.error("remote rpc request error", e);
            throw new IllegalArgumentException(e);
        }
        return new RpcResponse();
    }

    private void createChannel(ServiceProvider serviceProvider) {
        EventLoopGroup worker = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        RpcResponseHandler responseHandler = new RpcResponseHandler();
        bootstrap.group(worker)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ChannelPipeline pipeline = ch.pipeline();
                        //编码
                        pipeline.addLast("frameEncoder", new FrameEncoder());
                        pipeline.addLast("rpcRequestEncoder", new RpcRequestEncoder());
                        //解码
                        pipeline.addLast("frameDecoder", new FrameDecoder());
                        pipeline.addLast("rpcResponseDecoder", new RpcResponseDecoder());
                        //接收服务端
                        pipeline.addLast("rpcResponseHandler", responseHandler);
                    }
                });

        //建立连接
        try {
            ChannelFuture future = bootstrap.connect(serviceProvider.getServerIp(), serviceProvider.getRpcPort()).sync();
            //如果连接成功
            if (future.isSuccess()) {
                RpcRequestHolder.addChannelMapping(new ChannelMapping(serviceProvider.getServerIp(), serviceProvider.getRpcPort(), future.channel()));
                log.info("创建channel成功,serverIp{}, rpcPort:{}", serviceProvider.getServerIp(), serviceProvider.getRpcPort());
            }
        } catch (InterruptedException e) {
            log.info("客户端连接创建失败");
            e.printStackTrace();
        }
    }

}
